package kafka;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Properties;

/**
 * describe: 消费字符串的 kafka consumer
 *
 * @author helloworldyu
 * @data 2018/3/1
 */
public class StringKafkaConsumer {
	private Logger logger = LoggerFactory.getLogger(StringKafkaConsumer.class);
	Consumer<String, String> consumer;


	public StringKafkaConsumer(String brokerList, String groupId, Collection<String> topics, OFFSET offset) {
		Properties props = new Properties();

		//-----------------指定集群地址-----------------
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
		//-----------------指定客户端分组----------------
		props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		//-consumer所fetch的消息的offset将会自动的同步到zookeeper。这项提交的offset将在进程挂掉时，由新的consumer使用
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
		//-------设置自动提交 offset 到 zookeeper 的周期--------
		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
		//-------consumer 允许的最大数据 size------------------
		props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 20480000);
		//-------------设置 offset 起始位置--------------------
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offset.getOffset());
		//-------------------设置编解码器----------------------
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class);
		consumer = new KafkaConsumer<>(props);
		consumer.subscribe(topics);
		logger.debug("初始化 consumer: brokerList={}, groupId={}, topics={}", brokerList, groupId, topics);
	}

	/**
	 * earliest: automatically reset the offset to the earliest offset
	 * latest: automatically reset the offset to the latest offset
	 * none: throw exception to the consumer if no previous offset is found for the consumer's group
	 * anything else: throw exception to the consumer.
	 */
	public enum OFFSET {
		EARLIEST("earliest"),
		LATEST("latest"),
		NONE("none"),
		ANYTHING("anything"),;

		OFFSET(String offset) {
			this.offset = offset;
		}

		public String getOffset() {
			return offset;
		}

		private String offset;
	}

	public ConsumerRecords<String, String> poll(int timeout) {
		ConsumerRecords<String, String> records = consumer.poll(timeout);
		logger.debug(records.toString());
		return records;
	}

	public void close() {
		consumer.close();
	}
}